#ifndef __M_CHANNAL_H__
#define __M_CHANNAL_H__
#include "../mqCommon/logger.hpp"
#include "../mqCommon/helper.hpp"
#include "../mqCommon/mq_thread_pool.hpp"
#include "../mqCommon/mq_msg.pb.h"
#include "../mqCommon/mq_proto.pb.h"
#include "../thirdLib/net/TcpConnection.h"
#include "../thirdLib/proto/codec.h"
#include "../thirdLib/proto/dispatcher.h"
#include "mq_consumer.hpp"
#include <iostream>

namespace yqx{
    using ProtobufCodecPtr = std::shared_ptr<ProtobufCodec>;
    using basicConsumeResponsePtr = std::shared_ptr<basicConsumeResponse>;
    using basicCommonResponsePtr = std::shared_ptr<basicCommonResponse>;
    class Channel{
        public:
            using ptr = std::shared_ptr<Channel>;
            Channel(const muduo::net::TcpConnectionPtr& conn, const ProtobufCodecPtr& codec):
            _cid(UUIDHelper::uuid()), _conn(conn), _codec(codec)
            {}
            ~Channel(){
                if(_cmp != nullptr)
                    basicCancle();
            }
            std::string cid(){
                return _cid;
            }
            bool openChannel(){
                openChannelRequest req;
                std::string rid = UUIDHelper::uuid();
                req.set_rid(rid);
                req.set_cid(_cid);
                _codec->send(_conn, req);
                basicCommonResponsePtr resp = waitResponse(rid);
                return resp->ok();
            }
            void closeChannel(){
                closeChannelRequest req;
                std::string rid = UUIDHelper::uuid();
                req.set_rid(rid);
                req.set_cid(_cid);
                _codec->send(_conn, req);
                waitResponse(rid);
            }
            bool declareExchange(const std::string& ename, ExchangeType type, bool durable, bool auto_delete){
                //构造一个声明交换机的请求对象
                declareExchangeRequest req;
                req.set_rid(UUIDHelper::uuid());
                req.set_cid(_cid);
                req.set_exchange_name(ename);
                req.set_exchange_type(type);
                req.set_durable(durable);
                req.set_auto_delete(auto_delete);
                //向服务器发送请求
                _codec->send(_conn, req);
                //等待服务器相应
                basicCommonResponsePtr resp = waitResponse(req.rid());
                //返回
                return resp->ok();
            }
            void deleteExchange(const std::string& ename){
                deleteExchangeRequest req;
                req.set_rid(UUIDHelper::uuid());
                req.set_cid(_cid);
                req.set_exchange_name(ename);
                _codec->send(_conn, req);
                basicCommonResponsePtr resp = waitResponse(req.rid());
            }
            bool declareQueue(const std::string& qname, bool durable, bool exclusive, bool auto_delete){
                declareQueueRequest req; 
                req.set_rid(UUIDHelper::uuid());
                req.set_cid(_cid);
                req.set_queue_name(qname);
                req.set_durable(durable);
                req.set_exclusive(exclusive);
                req.set_auto_delete(auto_delete);
                _codec->send(_conn, req);
                basicCommonResponsePtr resp = waitResponse(req.rid());
                return resp->ok();
            }
            void deleteQueue(const std::string& qname){
                deleteQueueRequest req;
                req.set_rid(UUIDHelper::uuid());
                req.set_cid(_cid);
                req.set_queue_name(qname);
                _codec->send(_conn, req);
                basicCommonResponsePtr resp = waitResponse(req.rid());
                return;
            }
            bool bind(const std::string& ename, const std::string& qname, const std::string& binding_key){
                queueBindRequest req;
                req.set_rid(UUIDHelper::uuid());
                req.set_cid(_cid);
                req.set_exchange_name(ename);
                req.set_queue_name(qname);
                req.set_binding_key(binding_key);
                _codec->send(_conn, req);
                basicCommonResponsePtr resp = waitResponse(req.rid());
                return resp->ok();
            }
            bool unbind(const std::string& ename, const std::string& qname){
                queueUnbindRequest req;
                req.set_rid(UUIDHelper::uuid());
                req.set_cid(_cid);
                req.set_exchange_name(ename);
                req.set_queue_name(qname);
                _codec->send(_conn, req);
                basicCommonResponsePtr resp = waitResponse(req.rid());
                return resp->ok();
            }
            void basicPublish(const std:: string& ename, const BasicProperties* bp, const std::string& body){
                basicPublishRequest req;
                req.set_rid(UUIDHelper::uuid());
                req.set_cid(_cid);
                req.set_exchange_name(ename);
                req.set_body(body);
                if(bp){
                    req.mutable_properties()->set_id(bp->id());
                    req.mutable_properties()->set_delivery_mode(bp->delivery_mode());
                    req.mutable_properties()->set_routing_key(bp->routing_key());
                }
                _codec->send(_conn, req);
                basicCommonResponsePtr resp = waitResponse(req.rid());
                return;
            }
            void basicAck(const std::string& msg_id){
                basicAckRequest req;
                req.set_rid(UUIDHelper::uuid());
                req.set_cid(_cid);
                req.set_queue_name(_cmp->_qname);
                req.set_message_id(msg_id);
                _codec->send(_conn, req);
                basicCommonResponsePtr resp = waitResponse(req.rid());
                return;
            }
            void basicCancle(){
                basicCancelRequest req;
                req.set_rid(UUIDHelper::uuid());
                req.set_cid(_cid);
                req.set_queue_name(_cmp->_qname);
                req.set_consumer_tag(_cmp->_tag);
                _codec->send(_conn, req);
                basicCommonResponsePtr resp = waitResponse(req.rid());
                _cmp = nullptr;
                return;
            }
            bool basicConsume(const std::string& consumer_tag, const std::string& qname, bool auto_ack, const ConsumerCallback& cb){
                if(_cmp != nullptr){
                    LOG("当前信道已经订阅其他消息！");
                    return false;
                }
                basicConsumeRequest req;
                req.set_rid(UUIDHelper::uuid());
                req.set_cid(_cid);
                req.set_queue_name(qname);
                req.set_consumer_tag(consumer_tag);
                req.set_auto_ack(auto_ack);
                _codec->send(_conn, req);
                basicCommonResponsePtr resp = waitResponse(req.rid());
                if(resp->ok() == false){
                    LOG("订阅失败!");
                    return false;
                }
                _cmp = std::make_shared<Consumer>(consumer_tag, qname, auto_ack, cb);
                return true;
            }

        public:
            //连接收到基础相应后，向hash_map中添加响应
            void putBasicResponse(const basicCommonResponsePtr& resp){
                std::unique_lock<std::mutex> lock(_mtx);
                _basic_resp.insert(std::make_pair(resp->rid(), resp));
                _cv.notify_all();
            }
            //连接收到消息推送后，需要通过信道找到对应的消费者对象，通过回调函数进行消息处理
            void consume(const basicConsumeResponsePtr& resp){
                if(_cmp == nullptr){
                    LOG("消息处理时,未找到订阅者信息!");
                    return;
                }
                if(_cmp->_tag != resp->consumer_tag()){
                    LOG("收到的推送消息中的消费者标识与当前信道消费者表示不一致!");
                    return;
                }   
                _cmp->_callback(resp->consumer_tag(), resp->mutable_properties(), resp->body());
            }
        private:
            basicCommonResponsePtr waitResponse(const std::string& rid){
                std::unique_lock<std::mutex> lock(_mtx);
                _cv.wait(lock, [&rid, this](){
                    return _basic_resp.find(rid) != _basic_resp.end();
                });
                basicCommonResponsePtr basic_resp = _basic_resp[rid];
                _basic_resp.erase(rid);
                return basic_resp;
            }        
        private:
            std::string _cid;
            muduo::net::TcpConnectionPtr _conn;
            ProtobufCodecPtr _codec;
            Consumer::ptr _cmp;
            std::mutex _mtx;
            std::condition_variable _cv;
            std::unordered_map<std::string, basicCommonResponsePtr> _basic_resp;

    };
    class ChannelManager{
        public:
            using ptr = std::shared_ptr<ChannelManager>;
            ChannelManager(){}
            Channel::ptr create(const muduo::net::TcpConnectionPtr& conn, const ProtobufCodecPtr& codec){
                std::unique_lock<std::mutex> lock(_mtx);
                auto channel = std::make_shared<Channel>(conn, codec);
                _channels.insert(std::make_pair(channel->cid(), channel));
                return channel;
            }
            void remove(const std::string &cid){
                std::unique_lock<std::mutex> lock(_mtx);
                _channels.erase(cid);
            }
            Channel::ptr getChannel(const std::string &cid){
                std::unique_lock<std::mutex> lock(_mtx);
                auto it = _channels.find(cid);
                if(it == _channels.end()){
                    LOG("未找到信道");
                    return nullptr;
                }
                return it->second;
            }

        private:
            std::mutex _mtx;
            std::unordered_map<std::string, Channel::ptr> _channels;
    };
}   

#endif